package com.runa.monitor.platinterface.extend.amqpYiYuan;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import com.rabbitmq.client.impl.CredentialsProvider;
import com.rabbitmq.client.impl.DefaultCredentialsRefreshService;
import com.runa.monitor.comm.dto.newComm.AsynResult;
import com.runa.monitor.comm.dto.newComm.PanelVO;
import com.runa.monitor.comm.entity.CommPanel;
import com.runa.monitor.platinterface.mqtt.MqttUtil;
import com.runa.monitor.platinterface.platService.RepositoryService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.digest.DigestUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.concurrent.CountDownLatch;

/*
对接数据体
Received message: '{"createdAt":1700287962335,"data":{"packetId":"5","kv":{"temp_alarm":"2","quipment_number":"HX231117162932","account_number":"WG00B00C9D5F3F","maxhum_set":"100","temp_periodic_report":"3600","maxtemp_set":"100","minitemp_set":"-40","final_reporting_time":"1700287961000","temp_alarm_switch":false,"acquisition_frequency":"2","hum_alarm":"2","minihum_set":"0","temperature":"28.85595512390137","humidity":"24","hum_alarm_switch":false}},"deviceKey":"b00c9d5f3fcc","extData":{"occurredAt":"1700287961000"},"productKey":"p11rJT","subtype":"REPORT","ticket":"1175438438589005995","type":"MATTR"}', timestamp: 1700287963119
"kv":{"account_number":"WG00B00C9D5F3F","temperature":"25.57999992370606","humidity":"27"} 外加一个quipment_number
 */
@Service
@Slf4j
@Order(9999)
public class RabbitmqClientService implements CommandLineRunner {

    @Value("${yiYuan.enable:false}")
    private Boolean enable;

    @Value("${yiYuan.accessKey:24b9ygGwN549GVL6tiEzuvP1}")
    private String accessKey;

    @Value("${yiYuan.accessSecret:6AwhZZYooa8kwNmVsYcRsPqiNWuvrvJEB3PMAK4E}")
    private String accessSecret;

    @Value("${yiYuan.url:amqps://iot-amqp.quectelcn.com:5671/quec-open}")
    private String url;

    @Value("${yiYuan.queueName:a.1174363441445489.Saas}")
    private String queueName;

    @Resource
    private RepositoryService repositoryService;

    @Override
    public void run(String... args) throws Exception {
        if (!enable) return;
//        String accessKey = "24b9ygGwN549GVL6tiEzuvP1";
//        String accessSecret = "6AwhZZYooa8kwNmVsYcRsPqiNWuvrvJEB3PMAK4E";
//        String url = "amqps://iot-amqp.quectelcn.com:5671/quec-open", queueName = "a.1174363441445489.Saas";
        // 连接鉴权时用到的时间戳，断线重连时需要刷新timestamp的值为当前时间
        long timestamp = System.currentTimeMillis();
        String username = String.format("ver=1&auth_mode=accessKey&sign_method=sha256&access_key=%s&timestamp=%s", accessKey, timestamp);
        String password = DigestUtils.sha256Hex(username + accessSecret).toLowerCase();
        // AccessCredentials刷新周期，不能大于10分钟
        int RefreshTime = 4;
        try {
            CountDownLatch latch = new CountDownLatch(1);
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUri(url);
            factory.setUsername(username);
            factory.setPassword(password);
            factory.useSslProtocol();
            factory.setAutomaticRecoveryEnabled(true);
            factory.setTopologyRecoveryEnabled(true);
            factory.setCredentialsProvider(new AccessCredentialsProvider(accessKey, accessSecret, Duration.ofMinutes(RefreshTime)));
            factory.setCredentialsRefreshService(new DefaultCredentialsRefreshService.DefaultCredentialsRefreshServiceBuilder().build());
            try (Connection conn = factory.newConnection(); Channel channel = conn.createChannel()) {
                channel.basicQos(1); // DeliverCallback
                DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                    try {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8); // handle
//                        System.out.println("Received message: '" + message + "', timestamp: " + System.currentTimeMillis());
                        log.info("AMQPYiYuan Received message: '" + message);
                        JSONObject data = JSONObject.parseObject(message);
                        JSONObject kv = data.getJSONObject("data").getJSONObject("kv");
                        PanelVO panelVO = new PanelVO();
                        panelVO.setSerial(kv.getJSONObject("quipment_number").toString());
                        CommPanel panel = repositoryService.getCommPanelByNodeSerial(panelVO.getSerial());
                        if (null == panel) {
                            log.info("nb面板不存在，编号为：{}", panelVO.getSerial());
                            return;
                        }
                        panelVO.setTemperature(Float.valueOf(kv.getJSONObject("temperature").toString()));
                        panelVO.setHumidity(Float.valueOf(kv.getJSONObject("humidity").toString()));
                        panelVO.setSerial(kv.getJSONObject("account_number").toString());//户号
                        panelVO.setReadTime(new Date());
                        panelVO.setPanelDate(new Date());
                        AsynResult asynResult = new AsynResult();
                        asynResult.successSetting();
                        asynResult.setContent(Collections.singleton(panelVO));
                        asynResult.setIdentity(panel.getId());
                        MqttUtil.send(PanelVO.TOPIC, asynResult);
//                        JSONArray data = JSONObject.parseArray("rows");
                    } finally {
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    }
                }; // acutoAck true/false

                channel.basicConsume(queueName, false, deliverCallback, consumerTag -> {
                    System.out.println("The consumer is cancelled");
                    latch.countDown();
                });
                latch.await();
            }
        } catch (Exception e) {
            
        }
    }


    public static void main(String[] args) {
//        String accessKey = "${accessKey}";
//        String accessKey = "24b9yehUjg4hfpz719Yb9Mwu";
        String accessKey = "24b9ygGwN549GVL6tiEzuvP1";
//        String accessSecret = "${accessSecret}";
//        String accessSecret = "6AJFseQPg1F4uhdTUdiCDhkCyHwbzaFaLmuqG4uy";
        String accessSecret = "6AwhZZYooa8kwNmVsYcRsPqiNWuvrvJEB3PMAK4E";
//        String url = "${connectionUrl}", queueName = "${queueName}";
//        String url = "amqps://iot-amqp.quectelcn.com:5671/quec-open", queueName = "a.1169647340754926.ceshiduilie1";
//        String url = "amqpzs://iot-amqp.quectelcn.com:5671/quec-open", queueName = "a.1169647340754926.ceshiduilie1";
        String url = "amqps://iot-amqp.quectelcn.com:5671/quec-open", queueName = "a.1174363441445489.Saas";
        // 连接鉴权时用到的时间戳，断线重连时需要刷新timestamp的值为当前时间
        long timestamp = System.currentTimeMillis();
        String username = String.format("ver=1&auth_mode=accessKey&sign_method=sha256&access_key=%s&timestamp=%s", accessKey, timestamp);
        String password = DigestUtils.sha256Hex(username + accessSecret).toLowerCase();
        // AccessCredentials刷新周期，不能大于10分钟
        int RefreshTime = 4;

        try {
            CountDownLatch latch = new CountDownLatch(1);
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUri(url);
            factory.setUsername(username);
            factory.setPassword(password);
            factory.useSslProtocol();
            factory.setAutomaticRecoveryEnabled(true);
            factory.setTopologyRecoveryEnabled(true);
            factory.setCredentialsProvider(new AccessCredentialsProvider(accessKey, accessSecret, Duration.ofMinutes(RefreshTime)));
            factory.setCredentialsRefreshService(new DefaultCredentialsRefreshService.DefaultCredentialsRefreshServiceBuilder().build());

            try (Connection conn = factory.newConnection(); Channel channel = conn.createChannel()) {
                channel.basicQos(1); // DeliverCallback
                DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                    try {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8); // handle
                        System.out.println("Received message: '" + message + "', timestamp: " + System.currentTimeMillis());
                    } finally {
                        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    }
                }; // acutoAck true/false

                channel.basicConsume(queueName, false, deliverCallback, consumerTag -> {
                    System.out.println("The consumer is cancelled");
                    latch.countDown();
                });
                latch.await();
            }
        } catch (Exception e) {
            
        }
    }


}

/**
 * 用于刷新AccessCredentials，防止断线重连时AccessCredentials过期
 */
class AccessCredentialsProvider implements CredentialsProvider {

    private final String accessKey;
    private final String accessSecret;
    private final Duration expireDuration;

    private String username;
    private String password;

    public AccessCredentialsProvider(String accessKey, String accessSecret, Duration expireDuration) {
        this.accessKey = accessKey;
        this.accessSecret = accessSecret;
        this.expireDuration = expireDuration;
        refresh();
    }

    @Override
    public void refresh() {
        long timestamp = Calendar.getInstance().getTimeInMillis();
        this.username = String.format("ver=1&auth_mode=accessKey&sign_method=sha256&access_key=%s&timestamp=%s", accessKey, timestamp);
        this.password = DigestUtils.sha256Hex(username + accessSecret).toLowerCase();
    }

    @Override
    public String getUsername() {
        return this.username;
    }

    @Override
    public String getPassword() {
        return this.password;
    }

    @Override
    public Duration getTimeBeforeExpiration() {
        return this.expireDuration;
    }
}

